package com.athuahua.es;

import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.Result;
import co.elastic.clients.elasticsearch._types.query_dsl.MatchQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.indices.CreateIndexRequest;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.endpoints.BooleanResponse;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Stream;

/**
 * @Author: huahua
 * @Date: 2024/10/2 0002
 * @Time: 18:00
 * @Description:
 */
public class ESCoreMain {
    private static RestClient restClient;

    //构建ES服务器的传输对象
    private static ElasticsearchTransport transport;

    // 同步客户端对象 同步需要关闭传输对象
    private static ElasticsearchClient client;

    // 异步客户端对象
    private static ElasticsearchAsyncClient asyncClient;

    private static final String SERVER_URL = "http://192.168.37.129:9200";
    private static final String INDEX_NAME = "huahua";

    public static void main(String[] args) throws Exception {
        initESConnection();

        //operationIndex();
        //operationDocument();
        //queryDocument();
        //queryDocumentLambda();

        AsyncQueryDocument();
        //关闭客户端对象
        //restClient.close();
    }

    private static void initESConnection() throws Exception {
        //构建客户端对象
        restClient = RestClient.builder(
                HttpHost.create(SERVER_URL)
        ).build();

        //构建ES服务器的传输对象
        transport = new RestClientTransport(
                restClient, new JacksonJsonpMapper());

        // 同步客户端对象
        client = new ElasticsearchClient(transport);
        // 异步客户端对象
        asyncClient = new ElasticsearchAsyncClient(transport);
    }

    private static void operationIndex() throws Exception {

        //创建索引
//        CreateIndexRequest request = new CreateIndexRequest.Builder().index(INDEX_NAME).build();
//        client.indices().create(request);
        final boolean b = client.indices().exists(c -> c.index(INDEX_NAME)).value();
        if (b) {
            System.out.println("索引存在");
        } else {
            client.indices().create(c -> c.index(INDEX_NAME));
        }
        transport.close();

    }

    private static void operationDocument() throws Exception {
        User user = new User.UserBuilder().id(101).name("huahua").age(25).build();
        //创建文档
        Result result = client.create(req -> req
                        .index(INDEX_NAME)
                        .id(user.getId().toString())
                        .document(user))
                .result();
        System.out.println(result);

        //批量操作文档
        List<User> users = new ArrayList<>();
        users.add(new User.UserBuilder().id(102).name("huahua2").age(26).build());
        users.add(new User.UserBuilder().id(103).name("huahua3").age(27).build());
        users.add(new User.UserBuilder().id(104).name("huahua4").age(28).build());
        users.add(new User.UserBuilder().id(105).name("huahua5").age(29).build());

        client.bulk(req -> {
            users.forEach(u -> {
                req.operations(
                        b -> b.create(
                                d -> d
                                        .index(INDEX_NAME)
                                        .id(u.getId().toString())
                                        .document(u)
                        )
                );
            });
            return req;
        });


        transport.close();
    }

    private static void queryDocument() throws Exception {
        MatchQuery match = new MatchQuery.Builder()
                .field("age").query(25)
                .build();

        Query query = new Query.Builder()
                .match(match)
                .build();

        SearchRequest req = new SearchRequest.Builder()
                .query(query)
                .build();


        final SearchResponse<Object> search = client.search(req, Object.class);
        System.out.println(search);


        transport.close();
    }

    private static void queryDocumentLambda() throws Exception {
        SearchResponse<Object> search = client.search(
                r -> r.query(
                        q -> q.match(
                                m -> m.field("age").query(25)))
                , Object.class);
        System.out.println(search);


        transport.close();
    }

    private static void AsyncQueryDocument() throws Exception {
        asyncClient.indices().create(
                        req -> req.index("lalallalala")
                )
                .thenApply(//异步查询结果处理
                        res -> res.acknowledged()
                )
                .whenCompleteAsync( //当异步查询完成时
                        (res, err) -> {
                            if (res) {
                                System.out.println("回调方法" + res);
                            } else {
                                err.printStackTrace();
                            }
                        }
                );
        System.out.println("主线程代码...");
    }


}